/**
 * 
 */
package io.dora.remoting;

import io.dora.base.Protocol;
import io.dora.rpc.FutureResult;
import io.dora.rpc.RpcFutureResult;
import io.dora.rpc.RpcInvocation;
import io.dora.rpc.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * Consumer Netty 客户端实现
 * 
 * @author zhaohongbo
 *
 */
public class NettyConsumerClient implements ConsumerClient {
	
	private Logger logger = LoggerFactory.getLogger(NettyConsumerClient.class);
	
	/** Netty channel future  */
	private ChannelFuture channelFuture;
	
	/** 主机 */
	private String host;
	
	/** 端口 */
	private int port;
	
	private boolean success = true;
	
	/** 事件循环组 */
	private EventLoopGroup group;
	
	public NettyConsumerClient(String host, int port, Protocol protocol) {
		try {
			this.host = host;
			this.port = port;
			this.connect();
		} catch (Exception e) {
			logger.error("Consumer client init fail", e);
			setSuccess(false);
		}
	}

	@Override
	public void connect() {
		logger.info("Dora consumer client start connect");
        group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch)
                            throws Exception {
                        ch.pipeline().addLast(new ObjectEncoder());
                        ch.pipeline().addLast(new ObjectDecoder(1024 * 1024, ClassResolvers.cacheDisabled(null)));
                        ch.pipeline().addLast(new ConsumerHandler());
                    }
                });
        
        channelFuture = b.connect(host, port);
        channelFuture.awaitUninterruptibly();
        assert channelFuture.isDone();
    	
    	if (channelFuture.isCancelled()) {
            
        } else if (!channelFuture.isSuccess()) {
        	throw new RuntimeException(channelFuture.cause().getCause());
        }
	}
	
	public Channel getChannel() {
		return channelFuture.channel();
	}

	@Override
	public FutureResult request(RpcInvocation invocation) {
		RpcRequest request = new RpcRequest();
    	FutureResult futureResult = new RpcFutureResult(request.getId());
    	request.setData(invocation);
    	ChannelFuture f = channelFuture;
    	logger.info("Dora server channel: {}", f);
    	
    	if (!f.isSuccess()) {
    		throw new RuntimeException(channelFuture.cause().getCause());
        } else {
        	f.channel().writeAndFlush(request);
        }
        
    	return futureResult;
	}

	@Override
	public void close() {
		group.shutdownGracefully();
	}

	@Override
	public String getHost() {
		return this.host;
	}

	@Override
	public int getPort() {
		return this.port;
	}

	@Override
	public String getAddress() {
		return this.host + ":" + port;
	}

	@Override
	public boolean isSuccess() {
		return this.success;
	}

	@Override
	public void setSuccess(boolean success) {
		this.success = success;
	}
}
